package com.yungui.kafka.consumer.config;

import com.fasterxml.jackson.core.type.TypeReference;
import com.yungui.kafka.consumer.constant.KafkaConstant;
import com.yungui.kafka.consumer.entity.FlatMessage;
import com.yungui.kafka.consumer.util.JsonUtil;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.Deserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.math.BigDecimal;
import java.util.HashMap;
import java.util.Map;

/**
 * @author laiaiqin
 * @since 2019/5/6
 */
public class FlatMessageDeserializer implements Deserializer<FlatMessage> {
    private final static Logger LOG = LoggerFactory.getLogger(FlatMessageDeserializer.class);
    private final static Integer INT_DATA_TYPE = 1;
    private final static Integer LONG_DATA_TYPE = 2;
    private final static Integer DOUBLE_DATA_TYPE = 3;
    private final static String DEFAULT_TOPIC = "topic_robot";
    private String encoding = "UTF8";
    private TypeReference<FlatMessage> typeRef = new TypeReference<FlatMessage>() {
    };

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        String propertyName = isKey ? "key.deserializer.encoding" : "value.deserializer.encoding";
        Object encodingValue = configs.get(propertyName);
        if (encodingValue == null) {
            encodingValue = configs.get("deserializer.encoding");
        }

        if (encodingValue != null && encodingValue instanceof String) {
            this.encoding = (String) encodingValue;
        }
    }

    @Override
    public FlatMessage deserialize(String topic, byte[] data) {
        try {
            if (data == null) {
                return null;
            }
            String rawMessage = new String(data, this.encoding);
            FlatMessage flatMessage = JsonUtil.fromJson(rawMessage, typeRef);
            boolean needProcessDataType = false;
            if (DEFAULT_TOPIC.equals(topic)) {
                //默认topic，不需要处理删除语句的数据类型，因为不需要对外发出消息
                needProcessDataType = KafkaConstant.INDEX_EVENT_TYPES.contains(flatMessage.getType());
            } else {
                needProcessDataType = KafkaConstant.EVENT_TYPES.contains(flatMessage.getType());
            }
            if (needProcessDataType) {
                if(flatMessage.getMysqlType() == null || "mysql".equals(flatMessage.getDatabase())) {
                    return null;
                }
                processDataType(flatMessage);
            }
            return flatMessage;
        } catch (Exception e) {
            LOG.error("Error when deserializing byte[] to FlatMessage.", e);
            throw new SerializationException("Error when deserializing byte[] to FlatMessage.");
        }
    }

    /**
     * 数据类型转换：接收到之后值都是String, 需要根据mysql数据类型转换
     * tinyint,smallint,mediumint,int 转换成 Integer
     * bigint 转换成 Long
     * float,double,decimal 转换成 Double
     *
     * @param flatMessage
     * @return
     */
    private FlatMessage processDataType(FlatMessage flatMessage) {
        Map<String, Integer> colTypeMap = getNeedProcessColumns(flatMessage.getMysqlType());
        for (Map<String, Object> line : flatMessage.getData()) {
            for (Map.Entry<String, Object> entry : line.entrySet()) {
                String col = entry.getKey();
                Object value = entry.getValue();
                if (colTypeMap.containsKey(col) && value != null) {
                    line.put(col, convertDataType(colTypeMap.get(col), String.valueOf(value)));
                }
            }
        }
        return flatMessage;
    }

    /**
     * 值转换
     * @param dataType
     * @param value
     * @return
     */
    private Object convertDataType(Integer dataType, String value) {
        Object retValue = null;
        if (INT_DATA_TYPE.equals(dataType)) {
            retValue = Integer.valueOf(value);
        } else if (LONG_DATA_TYPE.equals(dataType)) {
            retValue = Long.valueOf(value);
        } else if (DOUBLE_DATA_TYPE.equals(dataType)) {
            retValue = new BigDecimal(value).doubleValue();
        } else {
            retValue = value;
        }
        return retValue;
    }

    /**
     * 根据mysql字段类型定义，获取要处理的字段名和要处理成的数据类型
     *
     * @param mysqlType
     * @return
     */
    private Map<String, Integer> getNeedProcessColumns(Map<String, String> mysqlType) {
        Map<String, Integer> colTypeMap = new HashMap<>(16);
        for (Map.Entry<String, String> entry : mysqlType.entrySet()) {
            String columnName = entry.getKey();
            String dataType = entry.getValue().toLowerCase();
            if (dataType.contains("bigint")) {
                colTypeMap.put(columnName, LONG_DATA_TYPE);
            } else if (dataType.contains("int")) {
                colTypeMap.put(columnName, INT_DATA_TYPE);
            } else if (dataType.contains("float") || dataType.contains("double") || dataType.contains("decimal")) {
                colTypeMap.put(columnName, DOUBLE_DATA_TYPE);
            }
        }
        return colTypeMap;
    }

    @Override
    public void close() {

    }
}
